package com.easy.stock.spider.html;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Properties;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.http.DefaultHttpRequest;
import org.jboss.netty.handler.codec.http.HttpHeaders;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.jboss.netty.handler.codec.http.HttpRequest;
import org.jboss.netty.handler.codec.http.HttpVersion;
import com.easy.stock.spider.HttpClientPipelineFactory;
import com.easy.stock.spider.Spider;
import com.easy.stock.spider.SpiderTask;
import com.easy.stock.spider.SpiderTaskComparator;
import com.easy.stock.util.Constants;

public abstract class BaseHtmlSpider implements Spider {
	
	protected static Log LOG = LogFactory.getLog("spider");

	protected int connectTimeout=2000;
	protected int requestTimeout=2000;
	protected int maxParallelTaskNum=10; // 默认值
	protected int maxRetryNum=5;	// 默认值
	protected int queueSize=100000;	// 默认值
	protected Queue<SpiderTask> taskQueue = new PriorityQueue<SpiderTask>(1000, new SpiderTaskComparator());
	
	protected AtomicInteger parallelNum = new  AtomicInteger(0);
	
	/**
	 * 所有的子spider通过这个接口来往队列里添加抓取任务
	 * 
	 * @param task
	 */
	public void addTask(SpiderTask task) {
		synchronized(taskQueue) {
			if(taskQueue.size()<queueSize)
				taskQueue.offer(task);
		}
	}
	
	public void offTask(SpiderTask task) {
		synchronized(taskQueue) {
			taskQueue.offer(task);
		}
	}
	
	public SpiderTask pollTask() {
		synchronized(taskQueue) {
			return taskQueue.poll();
		}
	}
	
	protected abstract List<SpiderTask> prepareSeeds();
	
	public void setup(Properties props) {
		if(props.containsKey("maxParallelTaskNum"))
			maxParallelTaskNum = Integer.parseInt(props.getProperty("maxParallelTaskNum"));
		if(props.containsKey("maxRetryNum"))
			maxRetryNum = Integer.parseInt(props.getProperty("maxRetryNum"));
		if(props.containsKey("queueSize"))
			queueSize = Integer.parseInt(props.getProperty("queueSize"));
	}
	
	
	public int crawl() throws Exception {
		
		// step1, 准备初始种子
		List<SpiderTask> seeds = prepareSeeds();
		if(seeds==null || seeds.isEmpty()) {
			LOG.info("no seed tasks!");
			return Constants.SUCCESS;
		}
		
		for(SpiderTask seed : seeds) {
			taskQueue.offer(seed);
		}
		LOG.info("begin crawl for "+this.getClass().getSimpleName()+" , seed num="+seeds.size());

		// 防止内存泄露
		seeds.clear();
		seeds=null;
		
		// step2, 初始化netty
        ClientBootstrap bootstrap = new ClientBootstrap(
                new NioClientSocketChannelFactory(
                        Executors.newCachedThreadPool(),
                        Executors.newCachedThreadPool()));
        bootstrap.setPipelineFactory(new HttpClientPipelineFactory());
		
		
		Random random = new Random(System.currentTimeMillis());
		// step3, 循环处理抓取任务，直到所有任务结束
		while(!taskQueue.isEmpty()) {
			
			SpiderTask task = pollTask();
			switch(task.getStatus()) {
			
			case SUCCESS:
				// 删除已经成功的任务
				parallelNum.decrementAndGet();
				task.getChannel().close();
				LOG.info("success to crawl url : "+task.getUrl());
				break;
				
			case FAILED:
				{
					if(task.getRetryNum()>maxRetryNum) {
						// 删除已经多次失败的任务
						parallelNum.decrementAndGet();
						task.getChannel().close();
						LOG.error("fail to crawl url : "+task.getUrl());
						break;
					} else {
						// 重试失败的任务
						task.setRetryNum(task.getRetryNum()+1);
						task.setStatus(SpiderTask.STATUS.READY);
						parallelNum.decrementAndGet();
						task.getChannel().close();
						offTask(task);
						break;
					}
				}
			
			case READY:
				{
					if(parallelNum.get()>=maxParallelTaskNum) {
						Thread.sleep(500);
						offTask(task);
						break;
					}
					
					// 启动这个任务
					// 下面的代码，顺序不能调整，防止死锁
					parallelNum.incrementAndGet();
					task.setStatus(SpiderTask.STATUS.RUNNING);
					if(!prepareGetUrl(bootstrap, task)) {
						task.setStatus(SpiderTask.STATUS.FAILED);
						offTask(task);
					}
				
					break;
				}
				
			case RUNNING: 
				{
					Thread.sleep(500);
					offTask(task);
					if(random.nextInt(100)<=1)
						LOG.info("task queuesize="+taskQueue.size()+", parallelNum="+parallelNum.get());
				}
			default:
				break;
			}
			
			
		}
		
		LOG.info("finish crawl for "+this.getClass().getSimpleName());
        bootstrap.releaseExternalResources();
		// step4, 关闭链接
		return Constants.SUCCESS;
		
	}
	
	private boolean prepareGetUrl(ClientBootstrap bootstrap, SpiderTask task) {
		
		try {
			URI uri = new URI(task.getUrl());
			String host = uri.getHost() == null? "localhost" : uri.getHost();
			int port = uri.getPort();
			if (port == -1) {
			    port = 80;
			}
			
			ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port));
			future.getChannel().getPipeline().addLast("handler", task.getHandler());
			Channel channel = future.awaitUninterruptibly().getChannel();
			task.setChannel(channel);
	        if (!future.isSuccess()) {
	        	LOG.error("prepareGetUrl connect error, reason=", future.getCause());
	            return false;
	        }
			
			// Prepare the HTTP request.
	        HttpRequest request = new DefaultHttpRequest(
	                HttpVersion.HTTP_1_1, HttpMethod.GET, uri.toASCIIString());
	        request.setHeader(HttpHeaders.Names.HOST, host);
	        request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
	        request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP);
	        // Send the HTTP request.
	        channel.write(request);
			return true;
		} catch (Exception e) {
        	LOG.error("prepareGetUrl request error, reason=", e);
			return false;
		}
		
	}
	
	public Queue<SpiderTask> getTaskQueue() {
		return taskQueue;
	}

	public void setTaskQueue(Queue<SpiderTask> taskQueue) {
		this.taskQueue = taskQueue;
	}
	
	public void setMaxParallelTaskNum(int maxParallelTaskNum) {
		this.maxParallelTaskNum = maxParallelTaskNum;
	}

	public void setMaxRetryNum(int maxRetryNum) {
		this.maxRetryNum = maxRetryNum;
	}

	public void setQueueSize(int queueSize) {
		this.queueSize = queueSize;
	}
	
	public void setConnectTimeout(int connectTimeout) {
		this.connectTimeout = connectTimeout;
	}

	public void setRequestTimeout(int requestTimeout) {
		this.requestTimeout = requestTimeout;
	}

}
